package com.dc.service.impl;

import com.alibaba.fastjson.JSON;
import com.dc.common.ResponseResultVO;
import com.dc.model.Device;
import com.dc.ro.EsDeviceRO;
import com.dc.service.EsService;
import com.dc.util.SnowflakeIdWorker;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.dc.common.ResponseResultVO.*;

@Service
@Slf4j
public class EsServiceImpl implements EsService {

    @Resource
    private RestHighLevelClient client;

    @Override
    public ResponseResultVO addIndex(String index) {
        if (this.existsIndex(index)) {
            return failResult(index + "索引库已经存在!");
        }
        CreateIndexRequest request = new CreateIndexRequest(index);
        boolean flag = false;
        try {
            CreateIndexResponse createIndexResponse = client.indices().create(request, RequestOptions.DEFAULT);
            flag = createIndexResponse.isAcknowledged();
            if (flag) {
                System.out.println("创建索引库:" + index + "成功！");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
        return insertResult(flag ? 1 : 0);
    }

    @Override
    public ResponseResultVO deleteIndex(String index) {
        DeleteIndexRequest request = new DeleteIndexRequest(index);
        boolean flag = false;
        try {
            AcknowledgedResponse acknowledgedResponse = client.indices().delete(request, RequestOptions.DEFAULT);
            flag = acknowledgedResponse.isAcknowledged();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return deleteResult(flag ? 1 : 0);
    }

    @Override
    public ResponseResultVO addIndexData(String index) {
        Device device = new Device();
        device.setId(SnowflakeIdWorker.generateId());
        device.setDeviceStatus(0);
        device.setCreateTime(new Date());
        IndexRequest request = new IndexRequest(index);
        request.id(String.valueOf(device.getId()));
        request.source(JSON.toJSONString(device), XContentType.JSON);
        //异步添加
        try {
            ActionListener actionListener = new ActionListener() {
                @Override
                public void onResponse(Object o) {
                    log.info("回调--->添加成功");
                }
                @Override
                public void onFailure(Exception e) {
                    log.error("回调--->添加失败");
                }
            };
            client.indexAsync(request, RequestOptions.DEFAULT, actionListener);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return successResult("成功");
    }

    /**
     * 批量插入数据
     */
    @Override
    public ResponseResultVO batchIndexData(String index, List dataList) {
        try {
            //异步添加
            ActionListener actionListener = new ActionListener() {
                @Override
                public void onResponse(Object o) {
                    log.info("回调--->批量添加成功");
                }
                @Override
                public void onFailure(Exception e) {
                    log.error("回调--->批量添加失败");
                }
            };
            BulkRequest request = new BulkRequest();
            for (Object dto : dataList) {
                Map map = JSON.parseObject(JSON.toJSONString(dto), Map.class);
                request.add(new IndexRequest(index).id(String.valueOf(map.get("id"))).source(JSON.toJSONString(dto), XContentType.JSON));
            }
            client.bulkAsync(request, RequestOptions.DEFAULT, actionListener);
            return successResult("成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return failResult("失败");
    }


    @Override
    public ResponseResultVO conditionQueryIndexData(EsDeviceRO ro) {

        SearchRequest searchRequest = new SearchRequest(ro.getIndex());
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();

        //条件
        ro.getQuery().forEach((key, value) -> {
            TermQueryBuilder deviceStatus = QueryBuilders.termQuery(key, value);
            sourceBuilder.query(deviceStatus);
        });

        //查询开始-结束 。可以用来分页使用.设置查询的起始索引位置和数量：currentPage=0，pageSize= 5，代表从第1条开始，共返回5条文档数据
        sourceBuilder.from(1 == ro.getCurrentPage() ? 0 : (ro.getCurrentPage() - 1) * ro.getPageSize());
        sourceBuilder.size(ro.getPageSize());

        //设置一个可选的超时，控制允许搜索的时间。
        sourceBuilder.timeout(new TimeValue(60, TimeUnit.SECONDS));
        //排序
        sourceBuilder.sort(new FieldSortBuilder("createTime").order(SortOrder.DESC));
        searchRequest.source(sourceBuilder);
        List<Device> list = new ArrayList<>();
        try {
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            //处理搜索结果
            RestStatus restStatus = searchResponse.status();
            if (restStatus != RestStatus.OK) {
                System.out.println("搜索错误");
            }
            SearchHits hits = searchResponse.getHits();
            hits.forEach(item -> list.add(JSON.parseObject(item.getSourceAsString(), Device.class)));

        } catch (IOException e) {
            e.printStackTrace();
        }
        return successResult(list);
    }

    @Override
    public ResponseResultVO deleteIndexData(EsDeviceRO ro) {
        DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest(ro.getIndex());
        ro.getQuery().forEach((key, value) -> {
            deleteRequest.setQuery(new TermQueryBuilder(key, value));
        });
        //版本冲突时继续
        deleteRequest.setConflicts("proceed");
        try {
            client.deleteByQuery(deleteRequest, RequestOptions.DEFAULT);
            return successResult("成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
        return failResult("失败");
    }

    public boolean existsIndex(String index) {
        GetIndexRequest request = new GetIndexRequest(index);
        boolean exists = false;
        try {
            exists = client.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return exists;
    }
}
